package com.swak.zookeeper;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatch.State;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;

import com.swak.leader.ChangeListener;
import com.swak.leader.Leader;
import com.swak.utils.Lists;

@SuppressWarnings("unused")
public class ZookeeperLeader implements Leader, LeaderLatchListener {

	private final CuratorFramework client;
	private final String path;
	private final LeaderLatch leaderLatch;
	private AtomicBoolean state = new AtomicBoolean(false);
	private List<ChangeListener> listeners = Lists.newArrayList();

	public ZookeeperLeader(CuratorFramework client, String path) throws Exception {
		this(client, path, null);
	}

	public ZookeeperLeader(CuratorFramework client, String path, ChangeListener listener) throws Exception {
		this.client = client;
		this.path = path;
		this.addListener(listener);
		this.leaderLatch = new LeaderLatch(client, path);
		this.leaderLatch.addListener(this);
		this.leaderLatch.start();
	}

	@Override
	public void isLeader() {
		state.set(true);
		this.fireChangeListeners(true);
	}

	@Override
	public void notLeader() {
		state.set(false);
		this.fireChangeListeners(false);
	}

	@Override
	public boolean isMe() {
		return state.get();
	}

	private void fireChangeListeners(boolean isme) {
		for (Iterator<ChangeListener> iterator = listeners.iterator(); iterator.hasNext();) {
			ChangeListener changeListener = (ChangeListener) iterator.next();
			changeListener.leaderChange(isme);
		}
	}

	@Override
	public void addListener(ChangeListener listener) {
		if (listener != null) {
			this.listeners.add(listener);
		}
	}

	@Override
	public void join() throws Exception {
		if (leaderLatch.getState() == State.CLOSED) {
			leaderLatch.start();
		}
	}

	@Override
	public void leave() throws IOException {
		if (leaderLatch.getState() != State.CLOSED) {
			leaderLatch.close();
		}
	}
}
